package com.shellyan.gmall.rt.util

import com.shellyan.gmall.rt.bean.AlertInfo
import io.searchbox.client.config.HttpClientConfig
import io.searchbox.client.{JestClient, JestClientFactory}
import io.searchbox.core.{Bulk, Index}
import org.apache.spark.rdd.RDD

import scala.collection.JavaConverters._

/**
 * @author Shelly An
 * @create 2020/9/9 10:25
 */
object ESUtil {

  val esUrls = Set(
    "http://hadoop102:9200", "http://hadoop103:9200", "http://hadoop104:9200"
  ).asJava
  val factory = new JestClientFactory
  val conf = new HttpClientConfig.Builder(esUrls)
    .maxTotalConnection(100)
    .connTimeout(1000 * 10)
    .readTimeout(1000 * 10)
    .build()

  factory.setHttpClientConfig(conf)


  //Iterator 非容器式的  用的时候才生成 mapPartitions的返回结果要求是这个
  def insertBulk(index: String, sources: Iterator[Object]) = {
    val client: JestClient = factory.getObject

    val builder = new Bulk.Builder()
      .defaultIndex(index)
      .defaultType("_doc")

    //    写法1：
    //    sources.foreach(source => {
    //        val action: Index = source match {
    //          case (id: String, s) =>
    //            new Index.Builder(s)
    //              .index(index)
    //              .build()
    //          case s =>
    //            new Index.Builder(s)
    //              .build()
    //        }
    //        builder.addAction(action)
    //      })

    //    写法2：
    //    sources.foreach{
    //      case (id: String, s) =>
    //        val action: Index = new Index.Builder(s)
    //          .id(id)
    //          .build()
    //        builder.addAction(action)
    //      case s =>
    //        val action: Index = new Index.Builder(s)
    //          .build()
    //        builder.addAction(action)
    //    }

    //优化
    sources.map {
      case (id: String, s) =>
        new Index.Builder(s)
          .id(id)
          .build()
      case s =>
        new Index.Builder(s)
          .build()
    }.foreach(builder.addAction)
//   .foreach(builder.addAction(_)) //部分应用函数多此一举

    client.execute(builder.build())
    client.close()
  }


  def insertSingle(index: String, source: Object) = {
    val client: JestClient = factory.getObject

    val action: Index = source match {
      case (id: String, s) =>
        new Index.Builder(s)
          .index(index)
          .`type`("_doc") //type在scala中是关键字
          .id(id)
          .build()
      case s =>
        new Index.Builder(s)
          .index(index)
          .`type`("_doc")
          .build()
    }

    /**
     * execute 同步：更安全
     * executeAsync 异步：更效率
     */
    client.execute(action)

    client.close()
  }

  //隐式转换类
  implicit class RichES(rdd:RDD[AlertInfo]){
    def saveToES(index: String): Unit = {
      rdd.foreachPartition(infoIt => {
        ESUtil
          .insertBulk("gmall_coupon_alert",
            infoIt.map(info => (s"${info.mid}_${info.ts / 1000 / 60}", info)))
      })
    }
  }
}

